package com.flink.examples.mysql;

import com.flink.examples.TUser;
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.sinks.TableSink;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description 将DataStream数据流插入到mysql表中
 * @Author JL
 * @Date 2020/09/17
 * @Version V1.0
 */
public class DataStreamSink {

    /**
     * 官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/jdbc.html
     */

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(2000);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //查询sql
        String sql = "insert into t_user (id,name,age,sex,address,createTimeSeries) values (?,?,?,?,?,?)";
        //封装数据
        TUser user = new TUser();
        user.setId(0);
        user.setName("zhao1");
        user.setAge(22);
        user.setSex(1);
        user.setAddress("CN");
        user.setCreateTimeSeries(System.currentTimeMillis());
        DataStream<TUser> sourceStream = env.fromElements(user);

        //从DataStream获取数据
//        Expression id = ExpressionParser.parseExpression("id");
//        Expression name = ExpressionParser.parseExpression("name");
//        Expression age = ExpressionParser.parseExpression("age");
//        Expression sex = ExpressionParser.parseExpression("sex");
//        Expression address = ExpressionParser.parseExpression("address");
//        Expression createTimeSeries = ExpressionParser.parseExpression("createTimeSeries");
//        Table table = tEnv.fromDataStream(sourceStream, id, name, age, sex, address, createTimeSeries );
        Table table = tEnv.fromDataStream(sourceStream,$("id"),$("name"),$("age"),$("sex"),$("address"),$("createTimeSeries"));

        //输出到mysql
        //设置表视图字段与类型
        TableSchema tableSchema = TableSchema.builder()
                .field("id", DataTypes.INT())
                .field("name", DataTypes.STRING())
                .field("age", DataTypes.INT())
                .field("sex", DataTypes.INT())
                .field("address", DataTypes.STRING())
                //.field("createTime", DataTypes.TIMESTAMP())
                .field("createTimeSeries", DataTypes.BIGINT())
                .build();

        //设置sink输出jdbc
        TableSink tableSink = JDBCAppendTableSink.builder()
                .setDrivername(MysqlConfig.DRIVER_CLASS)
                .setDBUrl(MysqlConfig.SOURCE_DRIVER_URL)
                .setUsername(MysqlConfig.SOURCE_USER)
                .setPassword(MysqlConfig.SOURCE_PASSWORD)
                .setQuery(sql)
                .setParameterTypes(tableSchema.getFieldTypes())
                .setBatchSize(100)
                .build();

        //将数据源注册到tableEnv视图result中
        tEnv.registerTableSink("result",
                tableSchema.getFieldNames(),
                tableSchema.getFieldTypes(),
                tableSink);
        //在指定的路径下注册，然后执行插入操作
        table.executeInsert("result");
    }
}
